go-micro v3添加熔断

docker-compose添加熔断服务

version: '3'
services:
  # 注册中心集群版本设置
  consul1:
    image: cap1573/consul
    container_name: node1
    # 以 server 模式启动
    command: agent -server -bootstrap-expect=3 -node=node1 -bind=0.0.0.0 -client=0.0.0.0 -datacenter=dc1
  consul2:
    image: cap1573/consul
    container_name: node2
    # 以 server 模式启动
    command: agent -server -retry-join=node1 -bootstrap-expect=3 -node=node2 -bind=0.0.0.0 -client=0.0.0.0 -datacenter=dc1
    depends_on:
      # 依赖consul1启动好之后
      - consul1
  consul3:
    image: cap1573/consul
    container_name: node3
    # 以 server 模式启动
    command: agent -server -retry-join=node1 -bootstrap-expect=3 -node=node3 -bind=0.0.0.0 -client=0.0.0.0 -datacenter=dc1
    depends_on:
      # 依赖consul1启动好之后
      - consul1
  # 添加对外暴露的节点 启动控制面板
  consul4:
    image: consul
    container_name: node4
    command: agent -retry-join=node1 -node=node4 -bind=0.0.0.0 -client=0.0.0.0 -datacenter=dc1 -ui
    ports:
      - 8500:8500
    depends_on:
      - consul2
      - consul3
  # 添加数据库
  paas-mysql:
    image: cap1573/mysql:5.6
    environment:
      MYSQL_ROOT_PASSWORD: 123456
      container_name: paas-mysql
    # 我们这里换个端口，本地已经有3306了
    ports:
      - "3308:3306"
    # 需要将重要的数据挂盘
    volumes:
      - ./mysql:/var/lib/mysql
  # 添加jaeger 服务
  jaeger:
    image: cap1573/jaeger
    ports:
      - "6831:6831/udp"
      - "16686:16686"
  # 添加熔断看板
  hystrix-dasboard:
    # 镜像名称
    image: cap1573/hystrix-dashboard
    ports:
      - "9002:9002"

然后重新启动

docker-compose up

看板访问地址：http://127.0.0.1:9002/hystrix

url一定要带上/hystrix，不然访问不到

base服务添加熔断器服务

go get github.com/afex/hystrix-go/hystrix

package main

import (
    "fmt"
    "github.com/wujie/base/handler"
    base "github.com/wujie/base/proto/base"
    "github.com/wujie/common"
    "github.com/afex/hystrix-go/hystrix"
    "github.com/asim/go-micro/v3/registry"
    "github.com/jinzhu/gorm"
    "github.com/opentracing/opentracing-go"
    "net"
    "net/http"
    //"github.com/afex/hystrix-go/hystrix"
    "github.com/asim/go-micro/plugins/registry/consul/v3"
    opentracing2 "github.com/asim/go-micro/plugins/wrapper/trace/opentracing/v3"
    "github.com/asim/go-micro/v3"
    _ "github.com/jinzhu/gorm/dialects/mysql"
)

func main() {
    //需要本地启动，mysql，consul中间件服务
    //1.注册中心
    //consul := consul.NewRegistry(func(options *registry.Options) {
    //  options.Addrs = []string{
    //      consulHost + ":" + strconv.FormatInt(consulPort, 10),
    //  }
    //})
    consul := consul.NewRegistry(func(options *registry.Options) {
        options.Addrs = []string{
            "localhost:8500",
        }
    })
    //2.配置中心，存放经常变动的变量
    consulConfig, err := common.GetConsulConfig("localhost", consulPort, "/micro/config")
    if err != nil {
        //common.Error(err)
        fmt.Println(err)
    }
    fmt.Println(consulConfig)
    //3.使用配置中心连接 mysql
    // 这里的path是相对于 /micro/config 目录路径
    mysqlInfo := common.GetMysqlFromConsul(consulConfig, "mysql")
    //初始化数据库
    fmt.Println(mysqlInfo)
    db, err := gorm.Open("mysql", mysqlInfo.User+":"+mysqlInfo.Pwd+"@("+mysqlInfo.Host+":3308)/"+mysqlInfo.Database+"?charset=utf8&parseTime=True&loc=Local")
    if err != nil {
        //命令行输出下，方便查看错误
        fmt.Println(err)
        common.Fatal(err)
        fmt.Println("mysql连接失败")
    }
    fmt.Println("mysql连接成功")
    defer db.Close()
    //禁止复表
    db.SingularTable(true)

    //4.添加链路追踪
    //t, io, err := common.NewTracer("go.micro.service.base", tracerHost+":"+strconv.Itoa(tracerPort))
    t, io, err := common.NewTracer("base", "localhost:6831")
    if err != nil {
        //common.Error(err)
        fmt.Println(err)
    }
    defer io.Close()
    // 设置全局变量
    opentracing.SetGlobalTracer(t)

    // 5. 添加熔断器，作为客户端需要启用
    hystrixStreamHandler := hystrix.NewStreamHandler()
    // 启动
    hystrixStreamHandler.Start()

    //启动监听程序
    go func() {
        // 需要使用对应的主机ip不用127.0.0.1
        //http://192.168.0.112:9092/turbine/turbine.stream
        //看板访问地址 http://127.0.0.1:9002/hystrix，url后面一定要带 /hystrix
        err = http.ListenAndServe(net.JoinHostPort("0.0.0.0", "9092"), hystrixStreamHandler)
        if err != nil {
            // 简单处理error
            fmt.Println(err)
        }
    }()

    service := micro.NewService(
        micro.Name("base-wj"),
        micro.Version("latest"),
        // 添加注册中心
        micro.Registry(consul),
        // 添加链路追踪
        micro.WrapHandler(opentracing2.NewHandlerWrapper(opentracing.GlobalTracer())),
        // 客户端的方式
        micro.WrapClient(opentracing2.NewClientWrapper(opentracing.GlobalTracer())),
    )

    service.Init()

    base.RegisterBaseHandler(service.Server(), new(handler.BaseHandler))

    // 启动服务
    if err := service.Run(); err != nil {
        //输出启动失败信息
        common.Fatal(err)
    }
}

把熔断注册到服务中去，我们还需要单独写一个hystrix的插件

go get github.com/asim/go-micro/v3/client

plugin/hystrix/hytrix.go

package hystrix

import (
    "context"
    "fmt"
    "github.com/afex/hystrix-go/hystrix"
    "github.com/asim/go-micro/v3/client"
)

type clientWrapper struct {
    client.Client
}

// Call 熔断逻辑
func (c *clientWrapper) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
    return hystrix.Do(req.Service()+"."+req.Endpoint(), func() error {
        //正常执行
        fmt.Println(req.Service() + "." + req.Endpoint())
        return c.Client.Call(ctx, req, rsp, opts...)
    }, func(e error) error {
        //走熔断逻辑,每个服务都不一样
        fmt.Println(req.Service() + "." + req.Endpoint() + "的熔断逻辑")
        return e
    })
}

func NewClientHystrixWrapper() client.Wrapper {
    return func(i client.Client) client.Client {
        return &clientWrapper{i}
    }
}

服务添加代码

service := micro.NewService(
        micro.Name("base-wj"),
        micro.Version("latest"),
        // 添加注册中心
        micro.Registry(consul),
        // 添加链路追踪
        micro.WrapHandler(opentracing2.NewHandlerWrapper(opentracing.GlobalTracer())),
        // 客户端的方式
        micro.WrapClient(opentracing2.NewClientWrapper(opentracing.GlobalTracer())),
        // 只作为客户端的时候起作用
        micro.WrapClient(hystrix2.NewClientHystrixWrapper()),
        // 添加限流
        micro.WrapHandler(ratelimit.NewHandlerWrapper(1000)),
    )